package com.iwdnb.gkgz.application.service.impl;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.Collectors;

import cn.hutool.core.date.DateUtil;
import com.google.common.collect.Lists;
import com.iwdnb.bmnf.common.lock.DistributedLock;
import com.iwdnb.bmnf.common.lock.RedisLockUtils;
import com.iwdnb.bmnf.common.service.CacheService;
import com.iwdnb.bmnf.common.utils.BooleanUtils;
import com.iwdnb.gkgz.application.service.RealtimeStockService;
import com.iwdnb.gkgz.application.service.StockDataSource;
import com.iwdnb.gkgz.application.service.StockLimitUpService;
import com.iwdnb.gkgz.common.model.dto.Stock;
import com.iwdnb.gkgz.common.model.dto.StockDayData;
import com.iwdnb.gkgz.common.model.dto.StockLimitUpData;
import com.iwdnb.gkgz.common.utils.RealtimeStockUtils;
import com.iwdnb.gkgz.common.utils.StockUtils;
import com.iwdnb.gkgz.common.utils.WorkDayUtils;
import com.iwdnb.gkgz.infrastructure.dao.model.query.StockLimitUpQuery;
import com.iwdnb.gkgz.infrastructure.dao.repo.StockLimitUpRepository;
import com.iwdnb.gkgz.infrastructure.dao.repo.StockRepository;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RLock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
@Slf4j
public class RealtimeStockServiceImpl implements RealtimeStockService {

    @Autowired
    StockDataSource dataSource;

    @Autowired
    private Executor executor;
    @Autowired
    private CacheService cacheService;
    @Autowired
    StockRepository stockRepository;
    @Autowired
    StockLimitUpRepository stockLimitUpRepository;
    @Autowired
    private RedisLockUtils redisLockUtils;

    private static final AtomicInteger count = new AtomicInteger(0);

    //@DistributedLock(key = "'syncRealtimeData'")
    @Override
    public void syncRealtimeData(List<String> codes) {
        //RLock lock = null;
        try {
            //锁定5分钟
            //lock = redisLockUtils.tryLock("syncRealtimeData", 5, 300);
            List<String> storeCodes = null;
            if (CollectionUtils.isEmpty(codes)) {
                List<StockLimitUpData> limitUpDataList = stockLimitUpRepository.queryNeedSyncLimitUpList(new Date());
                if (CollectionUtils.isEmpty(limitUpDataList)) {
                    log.info("没有待实时同步的股票,return");
                    return;
                }
                codes = limitUpDataList.stream().map(StockLimitUpData::getCode).collect(Collectors.toList());
                storeCodes = limitUpDataList.stream().filter(t -> BooleanUtils.isTrue(t.getTldFlag()))
                    .map(StockLimitUpData::getCode).collect(Collectors.toList());

            }
            int size = codes.size();
            int partitionSize = 900;
            if (size <= partitionSize) {
                int index = count.incrementAndGet();
                log.info("{}-开始同步今日实时数据", index);
                RealtimeStockUtils.syncRealtimeDatas(codes);
                storeRealtimeDataToDb(storeCodes);
                log.info("{}-完成同步今日实时数据:{},持久化:{}", index, size, storeCodes);
            } else {
                List<List<String>> partitionedList = Lists.partition(codes, partitionSize);
                AtomicInteger count = new AtomicInteger(0);
                for (List<String> subCodes : partitionedList) {
                    executor.execute(() -> {
                        RealtimeStockUtils.syncRealtimeDatas(subCodes);
                        count.addAndGet(subCodes.size());
                        if (count.get() == size) {
                            log.info("完成同步今日实时数据:" + count.get());
                        }
                    });
                }
            }
        } catch (Exception e) {
            log.error("同步实时数据异常", e);
        } finally {
            //redisLockUtils.unlock(lock);
        }
    }

    @Override
    public StockDayData getRealtimeData(String code) {
        return RealtimeStockUtils.getRealtimeData(code);
    }

    @Override
    public void storeRealtimeDataToDb(List<String> codes) {
        if (CollectionUtils.isEmpty(codes)) {
            List<StockLimitUpData> stockLimitUpDataList = stockLimitUpRepository.queryNeedSyncLimitUpList(new Date());
            if (CollectionUtils.isEmpty(stockLimitUpDataList)) {
                log.info("没有待同步的股票数据");
                return;
            }
            codes = stockLimitUpDataList.stream().map(StockLimitUpData::getCode).collect(Collectors.toList());
        }

        List<Stock> stockList = stockRepository.findByCodes(codes);
        Map<String, Stock> stockMap = stockList.stream().collect(Collectors.toMap(Stock::getCode, Function.identity()));
        for (String code : codes) {
            Stock stock = stockMap.get(code);
            if (Objects.isNull(stock)) {
                log.info("{}股票不存在", code);
                continue;
            }
            StockDayData stockDayData = RealtimeStockUtils.getRealtimeData(code);
            if (Objects.isNull(stockDayData)) {
                log.info("{}股票实时数据为空", code);
                continue;
            }
            stock.setPrice(stockDayData.getClosePrice());
        }
        log.info("更新股票实时数据:" + StringUtils.join(codes, ","));
        stockRepository.saveAll(stockList);
    }
}
